#include <set>
#include <iostream>
using namespace std;

#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include "ack_package.h"
#include "ping_package.h"
#include "reset_package.h"
#include "connecting_package.h"
#include "connecting_confirmed_package.h"

#include "u_peer.h"
#include "u_peer_manager.h"

#define MAX_SYSTEM_PROTOCOL 10

class UPeerImp : public IIoHandler, public boost::enable_shared_from_this<UPeerImp>
{
	class SendingPackage
	{
	public:
		SendingPackage(IPackage::Ptr package, boost::posix_time::ptime sendingTime, unsigned int seq) : sendingTime_(sendingTime), initializeTime_(sendingTime), sendedCount_(1), buf_(new boost::asio::streambuf)
		{
			ostream os(buf_.get());
			IPackage::write(os, *package, seq);
		}
		~SendingPackage(){}
		
		boost::shared_ptr<boost::asio::streambuf> buf_;
		boost::posix_time::ptime sendingTime_;
		boost::posix_time::ptime initializeTime_;
		int sendedCount_;
		boost::shared_ptr<boost::asio::deadline_timer> timer_;
	};

	enum State
	{
		INITIALIZED,
		CONNECTING,
		CONNECTED
	};

	const udp::endpoint remoteAddr_;
	IPackageSender::Ptr packageSender_;
	typedef map<unsigned int, IPackage::Ptr> ReceivingQueue;
	typedef map<unsigned int, SendingPackage> SendingQueue;
	SendingQueue sendingQueue_;
	ReceivingQueue receivingQueue_;
	unsigned int sendingSeq_;
	unsigned int minimalReceivingSeq_;

	unsigned int lostCount_;
	unsigned int sendingCount_;
	unsigned int sendedCount_;
	unsigned int receivedCount_;
	unsigned int ackSendedCount_;
	unsigned int wastedSendingCount_;
	__int64 delayedTimeMills_;

	boost::posix_time::time_duration rtti_;
	boost::posix_time::time_duration deltaRage_;
	boost::posix_time::ptime lastUpdatedTime_;

	bool disposed_;
	volatile State state_;

	boost::asio::deadline_timer tickTimer_;

	static const boost::posix_time::time_duration DefaultRtti;
	static const boost::posix_time::time_duration TickTime;
	static const int MaxRetrySending;

	void onTimeOut(unsigned int packageSeq)
	{
		if(disposed_)
		{
			return;
		}
		auto itr = sendingQueue_.find(packageSeq);
		if(itr == sendingQueue_.end())
		{
			return;
		}

		++lostCount_;
		auto currentTime = boost::posix_time::microsec_clock::local_time();
		if(++((*itr).second.sendedCount_) > MaxRetrySending && currentTime - (*itr).second.initializeTime_ > TickTime * 10)
		{
			close(UPeer::OUTTIME);
		}
		else
		{
			(*itr).second.sendingTime_ = currentTime;
			deltaRage_ = (deltaRage_ * 5) / 4;
			packageSender_->send(remoteAddr_, (*itr).second.buf_->data());
			(*itr).second.timer_->expires_from_now(rtti_ + ((deltaRage_ * 6)/5));
			(*itr).second.timer_->async_wait(boost::bind(&UPeerImp::onTimeOut, shared_from_this(), packageSeq));
		}
	}

	void renewRttr(const boost::posix_time::time_duration& instanceRtti)
	{
		auto delta = instanceRtti - rtti_;
		deltaRage_ = ((deltaRage_ * 9) + (delta.is_negative() ? -delta : delta)) / 10;
		rtti_ = (rtti_ * 9 + instanceRtti) / 10;
	}

	void replyAckPackage(unsigned int packageSeq)
	{
		if(disposed_)
		{
			return;
		}
		++ackSendedCount_;
		IPackage::Ptr package = IPackage::Ptr(new AckPackage(packageSeq));
		packageSender_->send(remoteAddr_, package , 0);
	}

	void tick()
	{
		if(disposed_)
		{
			return;
		}

		if(state_ == CONNECTED)
		{
			if(boost::posix_time::microsec_clock::local_time() -
				lastUpdatedTime_ > TickTime)
			{
				_send(IPackage::Ptr(new PingPackage()));
			}
		}
		start();

		dump();
	}

	void dump()
	{
		cout << "------------" << endl;
		cout << "sended: " << sendingCount_ << endl;
		cout << "confirmed sending: " << sendedCount_ << endl;
		cout << "unconfirmed sending: " << sendingQueue_.size() << endl;
		cout << "total lost count: " << lostCount_ << endl;
		cout << "rtti: " << rtti_.total_milliseconds()  << endl;
		cout << "rtti delta rage: " << deltaRage_.total_milliseconds() << endl;
		cout << "received: " << receivedCount_ << endl;
		cout << "minimum receiving seq: " << minimalReceivingSeq_ << endl;
		cout << "wasted sending count: " << wastedSendingCount_ << endl;

		if(sendedCount_ > 0){
			cout << "avg package delay: " << (delayedTimeMills_ / sendedCount_) << endl;
		}
	}

	void scheduleTick()
	{
		if(!disposed_)
		{
			tickTimer_.expires_from_now(TickTime);
			tickTimer_.async_wait(boost::bind(&UPeerImp::tick, shared_from_this()));
		}
	}

	void _send(IPackage::Ptr package)
	{
		if(disposed_)
		{
			return;
		}
		boost::posix_time::ptime currentTime = boost::posix_time::microsec_clock::local_time();
		auto curSeq = sendingSeq_++;
		auto sendingPackage = UPeerImp::SendingPackage(package, currentTime, curSeq);
		sendingPackage.timer_ = boost::shared_ptr<boost::asio::deadline_timer>(new boost::asio::deadline_timer(packageSender_->ioService()));

		sendingQueue_.insert(SendingQueue::value_type(curSeq, sendingPackage));

		packageSender_->send(remoteAddr_, sendingPackage.buf_->data());
		sendingPackage.timer_->expires_from_now(rtti_ + ((deltaRage_ * 6)/5));
		sendingPackage.timer_->async_wait(boost::bind(&UPeerImp::onTimeOut, shared_from_this(), curSeq));

		++sendingCount_;
	}

public:

	UPeerImp(const udp::endpoint& remoteAddr, IPackageSender::Ptr packageSender): remoteAddr_(remoteAddr), 
		packageSender_(packageSender), 
		rtti_(DefaultRtti), 
		sendingSeq_(1), 
		disposed_(false),
		minimalReceivingSeq_(0),
		lostCount_(0),
		sendingCount_(0),
		sendedCount_(0),
		receivedCount_(0), 
		ackSendedCount_(0),
		delayedTimeMills_(0),
		deltaRage_(DefaultRtti), 
		state_(INITIALIZED),
		tickTimer_(packageSender->ioService()), wastedSendingCount_(0)
	{

	}
	~UPeerImp(){}

	inline void close(const UPeer::ShutdownReason& reason) 
	{
		disposed_ = true;
		tickTimer_.cancel();
		UPeerManager::getInstace()->onConnectionClosed(remoteAddr_, reason);
	}

	inline void close()
	{
		packageSender_->send(remoteAddr_, IPackage::Ptr(new ResetPackage()), 0);
		close(UPeer::INPURPOSE);
	}

	inline void start() 
	{
		if(disposed_)
		{
			return;
		}
		lastUpdatedTime_ = boost::posix_time::microsec_clock::local_time();
		scheduleTick();
	}

	inline void send(IPackage::Ptr package)
	{
		_send(package);
	}

	inline void send(IPackage* package)
	{
		auto p = IPackage::Ptr(package);
		packageSender_->ioService().post(boost::bind(&UPeerImp::_send, shared_from_this(), p));
	}

	inline void sendConnectingPackage()
	{
		send(new ConnectingPackage());
		state_ = CONNECTING;
	}

	inline bool isDisposed() const
	{
		return disposed_;
	}

	inline bool isEstablished() const {return state_ == CONNECTED;}

	inline void onConnectingReceived(UPeer::Ptr upeer, bool connectingSide = false)
	{
		if(state_ != CONNECTED)
		{
			state_ = CONNECTED;
			if(!connectingSide)
			{
				send(IPackage::Ptr(new ConnectingConfirmedPackage()));
			}
			UPeerManager::getInstace()->onConnectionEstablished(upeer, connectingSide);
		}
	}

	inline void onReset(bool byremote = false)
	{
		rtti_ = DefaultRtti;

		if(state_ == CONNECTING)
		{
			if(!byremote)
			{
				packageSender_->send(remoteAddr_, IPackage::Ptr(new ResetPackage()), 0);
			}
			cout << "on reset from :" <<remoteAddr_.address().to_string() << ":" << remoteAddr_.port() << endl;
			sendingQueue_.clear();
			send(IPackage::Ptr(new ConnectingPackage()));
		}
		else
		{
			close(UPeer::RESETED);
		}
	}
	inline const udp::endpoint getRemoteAddr() const {return remoteAddr_;}

	inline void confirmPackage(unsigned int packageSeq)
	{
		if(disposed_)
		{
			return;
		}
		if(sendingCount_ == 0)
		{
			disposed_ = true;
		}
		auto itr = sendingQueue_.find(packageSeq);
		if(itr == sendingQueue_.end())
		{
			++wastedSendingCount_;
			return;
		}
		auto currentTime = boost::posix_time::microsec_clock::local_time();
		lastUpdatedTime_ = currentTime;
		boost::posix_time::time_duration instanceRtti = currentTime - (*itr).second.sendingTime_;
		++sendedCount_;
		renewRttr(instanceRtti);
		delayedTimeMills_ += (currentTime - (*itr).second.initializeTime_).total_milliseconds();
		sendingQueue_.erase(itr);
	}

	inline bool onPackageReceived(IPackage::Ptr package, unsigned int packageSeq, UPeer::Ptr upeer)
	{
		if(packageSeq == 0)
		{
			return true;
		}

		if(disposed_)
		{
			return false;
		}

		if(state_ == CONNECTED)
		{
			if(package->messageID() == ConnectingPackage::messageID_)
			{
				packageSender_->send(remoteAddr_, IPackage::Ptr(new PingPackage()), 0);
				return false;
			}
		}
		else if(state_ == INITIALIZED)
		{
			if(package->messageID() != ConnectingPackage::messageID_)
			{
				packageSender_->send(remoteAddr_, IPackage::Ptr(new ResetPackage()), 0);
				close(UPeer::RESETED);
				return false;
			}
			minimalReceivingSeq_ = packageSeq - 1;
		}
		else if(state_ == CONNECTING)
		{
			if(package->messageID() != ConnectingConfirmedPackage::messageID_)
			{
				onReset(package->messageID() == ResetPackage::messageID_);
				return false;
			}
		}

		replyAckPackage(packageSeq);
		lastUpdatedTime_ = boost::posix_time::microsec_clock::local_time();

		if(packageSeq <= minimalReceivingSeq_)
		{
			return false;
		}
		auto itr = receivingQueue_.find(packageSeq);
		if(itr != receivingQueue_.end())
		{
			return false;
		}

		++receivedCount_;
		if(packageSeq == minimalReceivingSeq_ + 1)
		{
			try
			{
				package->handle(upeer);
			}catch(...)
			{
				close();
			}
			
			auto next = receivingQueue_.begin();
			auto old = packageSeq + 1;
			while(next != receivingQueue_.end() && (*next).first == old)
			{
				auto pack = (*next).second;
				if(pack)
				{
					try
					{
						pack->handle(upeer);
					}catch(...)
					{
						close();
					}
					
				}
				next = receivingQueue_.erase(next);
				++old;
			}
			minimalReceivingSeq_ = old - 1;
			return false;
		}
		else
		{
			if(package->messageID() <= MAX_SYSTEM_PROTOCOL)
			{
				receivingQueue_.insert(ReceivingQueue::value_type(packageSeq, IPackage::Ptr()));
				return true;
			}
			else
			{
				receivingQueue_.insert(ReceivingQueue::value_type(packageSeq, package));
				return false;
			}
		}
	}
};

const boost::posix_time::time_duration UPeerImp::DefaultRtti = boost::posix_time::millisec(50);
const boost::posix_time::time_duration UPeerImp::TickTime = boost::posix_time::seconds(2); 
const int UPeerImp::MaxRetrySending = 10;


UPeer::UPeer(const udp::endpoint& remoteAddr, IPackageSender::Ptr packageSender) : imp_(new UPeerImp(remoteAddr, packageSender))
{
}

UPeer::~UPeer()
{
}


void UPeer::start()
{
	imp_->start();
}

void UPeer::close(const UPeer::ShutdownReason& reason)
{
	imp_->close(reason);
}

void UPeer::close()
{
	imp_->close();
}

void UPeer::send(IPackage* package)
{
	imp_->send(	package);
}

void UPeer::send(IPackage::Ptr package)
{
	imp_->send(package);
}

void UPeer::sendConnectingPackage()
{
	imp_->sendConnectingPackage();
}

bool UPeer::isDisposed() const
{
	return imp_->isDisposed();
}

void UPeer::confirmPackage(unsigned int packageSeq)
{
	imp_->confirmPackage(packageSeq);
}

bool UPeer::onPackageReceived(IPackage::Ptr package, unsigned int packageSeq)
{
	return imp_->onPackageReceived(package, packageSeq, shared_from_this());
}

void UPeer::onConnectingReceived(bool connectingSide)
{
	return imp_->onConnectingReceived(shared_from_this(), connectingSide);
}

void UPeer::onReset(bool byremote)
{
	imp_->onReset(byremote);
}

bool UPeer::isEstablished() const
{
	return imp_->isEstablished();
}

const udp::endpoint UPeer::getRemoteAddr() const 
{
	return imp_->getRemoteAddr();
}